package com.tencent.sr.iris.activity.service.activity;

import com.alibaba.fastjson.JSON;
import com.tencent.sr.iris.activity.common.config.IrisThreadPoolConfig;
import com.tencent.sr.iris.activity.common.enums.IrisErrorCode;
import com.tencent.sr.iris.activity.common.enums.UserTaskStatusEnum;
import com.tencent.sr.iris.activity.common.util.ShardingIdUtil;
import com.tencent.sr.iris.activity.repository.entity.TIrisActivityAwardRecordDO;
import com.tencent.sr.iris.activity.repository.entity.TIrisUserTaskRecordDO;
import com.tencent.sr.iris.activity.repository.mapper.TIrisUserTaskRecordMapper;
import com.tencent.sr.iris.activity.repository.mapper.example.TIrisUserTaskRecordExample;
import com.tencent.sr.iris.activity.service.config.ActivityConfig;
import com.tencent.sr.rmall.common.exception.retail.TRetailErrorException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class AwardMonitorService implements CommandLineRunner {

    @Resource
    private ActivityConfig activityConfig;

    @Value("${iris.activity.award.delay.queue:iris:activity:award:delay:queue}")
    private String delayQueueName;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private TIrisUserTaskRecordMapper userTaskRecordMapper;

    @Resource(name = IrisThreadPoolConfig.IRIS_AWARD_MONITOR_THREAD_EXECUTOR)
    private Executor irisAwardMonitorThreadExecutor;

    @Override
    public void run(String... args) {
        startConsumer();
    }

    public void startConsumer() {
        if (activityConfig.isAwardQueueStop()) {
            log.info("发奖监控队列开启配置关闭，请开启配置以开启队列");
            return;
        }

        CompletableFuture.runAsync(() -> {
            while (!activityConfig.isAwardQueueStop()) {
                try {
                    RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(delayQueueName);
                    redissonClient.getDelayedQueue(blockingQueue);
                    String msg = blockingQueue.poll(1, TimeUnit.MINUTES);
                    //收到消息进行处理
                    doConsume(msg);
                } catch (Exception e) {
                    log.error("发奖监控队列出错：", e);
                }
            }
        }, irisAwardMonitorThreadExecutor);
        log.info("启动发奖监控队列消费线程成功");
    }

    /**
     * 处理消息
     *
     * @param msg
     */
    private void doConsume(String msg) {
        TIrisActivityAwardRecordDO awardRecord = JSON.parseObject(msg, TIrisActivityAwardRecordDO.class);
        if (awardRecord == null) {
            return;
        }

        log.info("处理发奖信息：{}", msg);
        Integer shardingId = ShardingIdUtil.sharding(awardRecord.getUid());
        String taskRecordId = awardRecord.getTaskRecordId();
        TIrisUserTaskRecordExample example = new TIrisUserTaskRecordExample();
        example.createCriteria()
                .andPropertyEqualTo(TIrisUserTaskRecordExample.SHARDING_ID, shardingId)
                .andPropertyEqualTo(TIrisUserTaskRecordExample.BIZ_ID, taskRecordId);
        List<TIrisUserTaskRecordDO> taskRecords = userTaskRecordMapper.limitSelectByExample(example, 0, 1);
        if (CollectionUtils.isEmpty(taskRecords)) {
            String m = String.format("查询不到发奖信息对应的任务信息，发奖信息：{}", JSON.toJSONString(awardRecord));
            throw new TRetailErrorException(IrisErrorCode.AWARD_MONITOR_NOT_RECORD, m);
        }

        TIrisUserTaskRecordDO taskRecord = taskRecords.get(0);
        if (!Objects.equals(taskRecord.getUserTaskStatus(), UserTaskStatusEnum.FINISHED.getCode())) {
            String m = String.format("已发奖的任务状态不为已完成，发奖信息：{}，任务信息：{}",
                    JSON.toJSONString(awardRecord), JSON.toJSONString(taskRecord));
            throw new TRetailErrorException(IrisErrorCode.AWARD_MONITOR_RECORD_NOT_FINISH, m);
        }
    }

    /**
     * 发送消息
     *
     * @param awardRecord
     */
    public void produce(TIrisActivityAwardRecordDO awardRecord) {
        if (awardRecord == null) {
            return;
        }

        try {
            RBlockingQueue<String> blockingQueue = redissonClient.getBlockingQueue(delayQueueName);
            RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
            delayedQueue.offer(JSON.toJSONString(awardRecord), activityConfig.getAwardQueueDelaySeconds(),
                    TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("发送发奖监控队列失败：", e);
        }
    }
}
